package mysql

import java.util.concurrent.atomic.AtomicInteger
import javax.sql.DataSource

import model.JdbcRepo
import wangzx.scala_commons.sql._

/**
  *
  * 描述:
  *
  * @author hz.lei
  * @date 2018年03月28日 下午8:39
  */
class MysqlLockTest(val name: String) {
  /**
    *
    */
  var dataSource: DataSource = JdbcRepo.dataSource


  // 消息总条数计数器
  val counter = new AtomicInteger(0)
  // 批量处理, 每次从数据库取出消息的最大数量(window)
  val window = 100
  // 单轮处理的消息计数器, 用于控制循环退出.
  val resultSetCounter = new AtomicInteger(0)


  def business(): Unit ={
    while(true){
      (1 to 100).foreach { i =>
        dataSource.executeUpdate("insert into common_event set event_type = 'sku'")
      }
      Thread.sleep(10)
    }
  }

  /**
    *
    */
  def test(): Unit = {
    do  {
      resultSetCounter.set(0)
      dataSource.withTransaction(conn => {
        val res = conn.row[Row](sql"""SELECT * FROM event_lock WHERE id = 1 FOR UPDATE """)
        val now = System.currentTimeMillis()
        println(f"$now%tT $name takes the lock")
//        Thread.sleep(1000)
        //val lockId = insert into xxx
        conn.eachRow[EventStore](sql"SELECT * FROM common_event limit ${window}")(event => {
          val result: Int = conn.executeUpdate(sql"DELETE FROM common_event WHERE id = ${event.id}")

          if (result == 1) {
//            println("ready to send !!!")
            counter.incrementAndGet()
          }
          resultSetCounter.incrementAndGet()

        })
        println(s"$name process ${resultSetCounter.get} rows\n")
      })
    } while(resultSetCounter.get() == window)

    // println(s" $name end publish messages(${counter.get()}) to kafka")

  }


}

case class EventStore(id: Long, eventType: String, eventBinary: Array[Byte])
